package com.lagou.hbase.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;

public class HbaseCreate {

    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() throws IOException {
        //获取一个配置文件对象
        conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum","server1,server2");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        //通过conf获取到hbase集群的连接
        conn = ConnectionFactory.createConnection(conf);
    }

    //创建一张hbase表
    @Test
    public void createTable() throws IOException {
        //获取HbaseAdmin对象用来创建表
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        //创建Htabledesc描述器，表描述器
        HTableDescriptor relations = new HTableDescriptor(TableName.valueOf("relations"));
        //指定列族
        relations.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(relations);
        System.out.println("relations表创建成功！！");
    }

    //向relations表中添加数据
    @Test
    public void putData() throws IOException {
        //准备table对象
        Table relations = conn.getTable(TableName.valueOf("relations"));

        //准备put对象
        Put uid1 = new Put(Bytes.toBytes("uid1"));
        uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("2"), Bytes.toBytes("2"));
        uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("5"), Bytes.toBytes("5"));
        uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("7"), Bytes.toBytes("7"));

        Put uid2 = new Put(Bytes.toBytes("uid2"));
        uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("1"), Bytes.toBytes("1"));
        uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("3"), Bytes.toBytes("3"));
        uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("6"), Bytes.toBytes("6"));
        uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("10"), Bytes.toBytes("10"));

        Put uid3 = new Put(Bytes.toBytes("uid3"));
        uid3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("1"), Bytes.toBytes("1"));

        Put uid4 = new Put(Bytes.toBytes("uid4"));
        uid4.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("2"), Bytes.toBytes("2"));

        //准备list，批量插入数据
        ArrayList<Put> puts = new ArrayList<Put>();
        puts.add(uid1);
        puts.add(uid2);
        puts.add(uid3);
        puts.add(uid4);

        relations.put(puts);
        System.out.println("插入数据成功");
    }

    //获取relations表数据
    @Test
    public void scanData() throws IOException {

        //准备table对象
        Table relations = conn.getTable(TableName.valueOf("relations"));
        //准备scan对象
        Scan scan = new Scan();
        //执行扫描
        ResultScanner resultScanner = relations.getScanner(scan);

        for (Result result : resultScanner) {
            //获取result中所有cell对象，遍历打印
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println("rowkey=>" + rowkey + "\t" + "family=>" + family + "\t" +
                        "column=>" + column + "\t" + "value=>" + value);
            }
            System.out.println();
        }
        relations.close();
    }

    //释放连接
    @After
    public void realse() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
